﻿using RockBus.Configuration.Interfaces;
using RockBus.DataContracts;
using RockBus.Logging.Interfaces;
using RockBus.Messages.Interfaces;
using RockBus.PerformanceCounters.Interfaces;
using RockBus.Routing.Interfaces;
using System.Reflection;
using System.Transactions;

namespace RockBus.MessageProcessors.MessageRepository
{
    public class MessageProcessor : IMessageProcessor
    {
        private const string OperationCounterName = "PersistMessage";

        public MessageProcessor(
            ILogger logger,
            IReadOnlyConfiguration configuration,
            IPerformanceCounterFactory performanceCounterFactory,
            IMessageFactory messageFactory,
            IQueueManager queueManager,
            IMessageRepositoryHelper messageRepositoryHelper)
        {
            this.Logger = logger;
            this.Configuration = configuration;
            this.PerformanceCounterFactory = performanceCounterFactory;
            this.MessageFactory = messageFactory;
            this.QueueManager = queueManager;
            this.MessageRepositoryHelper = messageRepositoryHelper;
        }

        private ILogger Logger { get; set; }

        private IReadOnlyConfiguration Configuration { get; set; }

        private IPerformanceCounterFactory PerformanceCounterFactory { get; set; }

        private IMessageFactory MessageFactory { get; set; }

        private IQueueManager QueueManager { get; set; }

        private IMessageRepositoryHelper MessageRepositoryHelper { get; set; }

        public MessageProcessorEnum ProcessingStep { get { return MessageProcessorEnum.PersistenceSendPort; } }

        public void ProcessMessage(IMessageContext messageContext)
        {
            this.Logger.LogDebug(MethodBase.GetCurrentMethod(), "{0}", messageContext);

            // Complete the request (or response) queue entry
            long? queueEntryId = null;
            object queueEntryIdObj = null;
            if (messageContext.GetProperty(MessagePublisher.QueueEntryIdKey, out queueEntryIdObj))
            {
                queueEntryId = (long)queueEntryIdObj;
            }

            if (messageContext.Exception != null)
            {
                // The PersistenceSendPorts are responsible for aborting the message
                messageContext.Abort();

                if (queueEntryId.HasValue)
                {
                    this.MessageRepositoryHelper.RollbackEntry(queueEntryId.Value, messageContext.Exception);
                }
                return;
            }
            else if (messageContext.ProcessingData.MessageDirection == MessageDirectionEnum.End)
            {
                if (queueEntryId.HasValue)
                {
                    using (var ts = messageContext.CreateTransactionScope(TransactionScopeOption.Required, this.Configuration.InternalConfiguration.IsolationLevel, this.Configuration.InternalConfiguration.TransactionTimeoutMs))
                    {
                        this.MessageRepositoryHelper.CompleteEntry(queueEntryId.Value);
                        ts.Complete();
                    }
                }
                return;
            }

            var oc = this.PerformanceCounterFactory.CreateOperationCounter(OperationCounterName);
            oc.Start();

            try
            {
                var md = this.MessageRepositoryHelper.StoreMessage(messageContext.Message, messageContext.ProcessingData.CorrelationId);

                foreach (var routingHeader in messageContext.RoutingHeaders)
                {
                    this.MessageRepositoryHelper.EnqueueEntry(md.messageId, routingHeader);
                }

                using (var ts = messageContext.CreateTransactionScope(TransactionScopeOption.Required, this.Configuration.InternalConfiguration.IsolationLevel, this.Configuration.InternalConfiguration.TransactionTimeoutMs))
                {
                    this.MessageRepositoryHelper.CompleteMessage(md);

                    if (messageContext.ProcessingData.IsResponse)
                    {
                        // Complete the request queue entry
                        if (queueEntryId.HasValue)
                        {
                            this.MessageRepositoryHelper.CompleteEntry(queueEntryId.Value);
                        }
                    }

                    ts.Complete();
                }

                messageContext.ProcessingData.MessageDirection = MessageDirectionEnum.End;
                this.QueueManager.ProcessMessage(messageContext);
            }
            catch
            {
                oc.Abort();
                throw;
            }
            finally
            {
                oc.Complete();
            }
        }
    }
}